package com.windy.kafka.trace.consumer;

import com.windy.common.transmit.kafka.KafkaConsumerTransitHeader;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.common.serialization.Deserializer;
import org.springframework.kafka.core.ConsumerFactory;

import java.util.Map;

/**
 * Created by liyufeng on 2022/5/8
 */
@Deprecated
public class TracingConsumerFactory<K, V> implements ConsumerFactory<K, V> {

    private final ConsumerFactory<K, V> consumerFactory;

    private KafkaConsumerTransitHeader kafkaConsumerTransitHeader;

    public TracingConsumerFactory(ConsumerFactory<K, V> consumerFactory,KafkaConsumerTransitHeader kafkaConsumerTransitHeader) {
        this.consumerFactory = consumerFactory;
        this.kafkaConsumerTransitHeader=kafkaConsumerTransitHeader;
    }

    @Override
    public Consumer<K, V> createConsumer() {
        return new TracingkakfaConsumer<>(consumerFactory.createConsumer(),kafkaConsumerTransitHeader);
    }

    @Override
    public Consumer<K, V> createConsumer(String clientIdSuffix) {
        return new TracingkakfaConsumer<>(consumerFactory.createConsumer(clientIdSuffix),kafkaConsumerTransitHeader);
    }

    @Override
    public Consumer<K, V> createConsumer(String groupId, String clientIdSuffix) {
        return new TracingkakfaConsumer<>(consumerFactory.createConsumer(groupId, clientIdSuffix),kafkaConsumerTransitHeader);
    }

    @Override
    public Consumer<K, V> createConsumer(String groupId, String clientIdPrefix,
                                         String clientIdSuffix) {
        return new TracingkakfaConsumer<>(
                consumerFactory.createConsumer(groupId, clientIdPrefix, clientIdSuffix),kafkaConsumerTransitHeader);
    }

    @Override
    public boolean isAutoCommit() {
        return consumerFactory.isAutoCommit();
    }

    @Override
    public Map<String, Object> getConfigurationProperties() {
        return consumerFactory.getConfigurationProperties();
    }

    @Override
    public Deserializer<K> getKeyDeserializer() {
        return consumerFactory.getKeyDeserializer();
    }

    @Override
    public Deserializer<V> getValueDeserializer() {
        return consumerFactory.getValueDeserializer();
    }
}
